package com.example.bigdata.spark.SparkTest

import org.apache.log4j.{Level, Logger}
import org.apache.spark.{SparkConf, SparkContext}

object ActionsDemo {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[1]").setAppName(TransformationDemo.getClass.getSimpleName)
    Logger.getLogger("org.apache.spark").setLevel(Level.OFF)
    val sc = new SparkContext(conf)
    //actionReduce(sc)
    //actionCollect(sc)
    //actionCount(sc)
    //actionTake(sc)
    //actionFirst(sc)
    //actionSaveAsFile(sc)
    actionForeach(sc)
  }

  /**
    * Reduce 通过函数func聚集数据集中的所有元素。Func函数接受2个参数，返回一个值。
    * 这个函数必须是关联性的，确保可以被正确的并发执行。
    * 需要注意的是，不同于Transformation算子，其结果仍然是RDD，但是执行Actions算子之后，其结果不再是RDD，而是一个标量。
    */
  def actionReduce(sc:SparkContext): Unit ={
    val list = List(1, 2, 3, 4, 5, 6)
    val listRDD = sc.parallelize(list)
    val ret = listRDD.reduce((v1,v2) => v1 + v2 )
    println("ret : " + ret)
  }

  /**
    * Collect
    * 在Driver的程序中，以数组的形式，返回数据集的所有元素。这通常会在使用filter或者其它操作后，
    * 返回一个足够小的数据子集再使用，直接将整个RDD集Collect返回，很可能会让Driver程序OOM，这点尤其需要注意。
    */
  def actionCollect(sc:SparkContext): Unit ={
    val list = List(1,2,3,4,5,6)
    val listRDD = sc.parallelize(list)
    val ret = listRDD.collect()
    ret.foreach(x => print(x +":"))
  }

  /**
    * Count
    * 返回数据集的元素个数。
    */
  def actionCount(sc:SparkContext): Unit ={
    val list = List(1, 2, 3, 4, 5, 6)
    val listRDD = sc.parallelize(list)
    val ret = listRDD.count()
    println("ret count : " + ret)
  }

  /**
    * Take
    *   返回一个数组，由数据集的前n个元素组成，这个操作目前并非在多个节点上，并行执行，而是Driver程序所在机器，
    *   单机计算所有的元素(Gateway的内存压力会增大，需要谨慎使用)
    */
  def actionTake(sc:SparkContext): Unit ={
    val list = List(1,2,3,4,5,6)
    val listRDD = sc.parallelize(list)
    val take = listRDD.take(3)
    take.foreach(x => print(x+","))
  }

  /**
    * First
    *    返回数据集的第一个元素（类似于take(1)）
    */
  def actionFirst(sc:SparkContext): Unit ={
    val list = List(10, 2, 3, 4, 5, 6)
    val listRDD = sc.parallelize(list)
    println(listRDD.first())
  }

  /**
    * SaveAsFile 将数据集的元素，以textfile的形式，保存到本地文件系统，hdfs或者任何其它hadoop支持的文件系统。
    * Spark将会调用每个元素的toString方法，并将它转换为文件中的一行文本。
    */
  def actionSaveAsFile(sc:SparkContext): Unit ={
    val list = List(1, 2, 3, 4, 5, 6)
    val listRDD = sc.parallelize(list)
    listRDD.saveAsTextFile("output/actionSaveAsFile")
  }

  /**
    * Foreach 在数据集的每一个元素上，运行函数func。这通常用于更新一个累加器变量，或者和外部存储系统做交互。
    *
    */
  def actionForeach(sc:SparkContext): Unit ={
    val list = List(1, 2, 3, 4, 5, 6)
    val listRDD = sc.parallelize(list)
    listRDD.foreach(println)
  }


  def actionSaveAsNewAPIHadoopFile(sc:SparkContext): Unit ={

  }

}
